package com.project.stream_consumer.demo11.controller;

import com.project.stream_consumer.demo7.conf.MyInPut;
import com.rabbitmq.client.Channel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;


@Component
@EnableBinding(MyInPut.class)
public class ConsunmerController {
    @KafkaListener(topics = MyInPut.MYKAFKAINPUT, containerFactory="batchFactory")
    public void consumerBatch(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) throws InterruptedException {
        System.out.println("接收到消息数量,通过BatchConsumerConfig-》configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 3)配置："+records.size());
        // TimeUnit.SECONDS.sleep(5);
        for(ConsumerRecord record: records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            //System.out.println("Received: " + record);
            if (kafkaMessage.isPresent()) {
                Object message = record.value();
                String topic = record.topic();
                System.out.println("接收到消息：" + message);
            }
        }
        ack.acknowledge();//确认消费
    }
}